OTAP Parquet study by jmacd · Pull Request #3405 · open-telemetry/otel-arrow

jmacd · 2026-07-01T23:30:32Z

Compares the cost of OTAP/IPC transport vs Parquet.

Then, compares a flattened schema compatible with Parquet that we might called OTAP-flat as opposed to OTAP-standard, which is denormalized. Points out that ~10 of the columns in the OTAP logs table convert directly into Parquet, so the cost of OTAP-to-Parquet-to-OTAP can be improved using zero-copy for the bulk of columns. Also considers whether OTAP-flat with RLE instead of materialized is competitive: it is, but it doesn't convert well to Parquet because Parquet is a quite-old format.

…rquet Adds a benchmark and unit-tested support library comparing the read and write cost, serialized size, and server-side OTAP-to-Parquet conversion cost of OTAP logs. It measures the current OTAP interleaved Arrow IPC representation against three flattened single-file Parquet layouts named nested, map, and wide, across the zstd, lz4, snappy, and none compressors. The server_cost model quantifies where the OTAP-to-Parquet conversion CPU lands when the data becomes Parquet on the server either way. It compares the server converting received OTAP/IPC against the server accepting Parquet the client already produced. This is dev-only benchmark tooling, so no changelog entry is included. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Adds a Vortex file-format contender behind the optional `vortex` cargo feature. It reuses the nested flattening, stores the flat Arrow batch in an in-memory Vortex file, then reads it back and unflattens to OTAP. Vortex 0.75 pins the same arrow-rs 58.3 as this workspace, so Arrow arrays interoperate directly and the round-trip runs fully in memory. Vortex 0.75 has no FixedSizeBinary encoding, so trace_id and span_id are cast to Binary before writing and restored on read. Decode targets an explicit plain Arrow schema so the OTAP schema check accepts the result. The Vortex path is gated so the default benchmarks build does not pull the heavy dependency. On this OTAP-logs full round-trip, Vortex at default settings is larger on the wire and much slower to write than zstd-Parquet. The benchmark README records the numbers and the caveats, including that Vortex targets selective reads that this workload does not exercise. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Adds a vortex-fast contender that writes the Vortex file with BtrBlocks disabled, to test whether skipping compression lets Vortex write faster than Parquet, and switches the input shapes to realistic 10k and 100k log-record blocks. The result is that Vortex does not write faster than Parquet here. Even with compression off, Vortex write is slower than Parquet write at both sizes, and disabling BtrBlocks also disables dictionary encoding, so the denormalized data fully expands to 13 MB at 10k and 186 MB at 100k. The write bottleneck is the Vortex write pipeline itself, which canonicalizes, repartitions, computes zone statistics, and builds a layout and footer, not the compressor. The README records the numbers and notes that Vortex targets selective reads this workload does not exercise. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Removes the Vortex contender and its optional dependency, and refines the otap_parquet study to break each pipeline into sub-steps at 10k, 50k, and 100k block sizes. The OTAP/IPC path is split into transport-optimize and Arrow IPC serialize on the encode side, and IPC deserialize and transport-decode on the decode side. The Parquet path is split into flatten and Parquet write on encode, and Parquet read and unflatten on decode. The benchmark prints an OTAP/IPC breakdown table and a Parquet breakdown table alongside the size table. The breakdown shows that IPC encode is dominated by the transport-optimized encoding, Parquet encode is dominated by the Parquet writer, and IPC is about an order of magnitude cheaper than Parquet on both the encode and decode side. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Adds a short analysis of the otap_parquet ratios, with lz4 included in the Parquet tables since that is the compressor used in the comparison elsewhere. The analysis explains that IPC encode is dominated by the transport-optimized encoding rather than compression, that IPC serialize is faster when compressed because there is less data to move, that Parquet encode is dominated by the writer with a large flatten tax, and that lz4 IPC decode is about six times slower than zstd in this Arrow IPC build. That last point shrinks the IPC-over-Parquet decode advantage from about 22x with zstd to under 5x with lz4, so measurements must hold the compressor fixed. The lz4 rows are also added to the README Parquet breakdown table. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

The single-batch size comparison left the Arrow IPC streaming benefit off the table, so this adds a streaming measurement and documents the effect. A long-lived Producer writes the Arrow schema once and delta-encodes dictionaries, so every batch after the first omits the schema and re-sends only new dictionary entries. The measured amortization is a fixed cost of about 11 KB per batch. For small frequent batches this flips the size verdict, because at 1000 records the steady-state IPC batch is 0.69x the Parquet file, and Parquet has no equivalent per-batch amortization. ipc::stream_batch_sizes measures this and the bench prints a streaming table. This also corrects the block sizes. A single OTAP logs batch holds at most 65535 records because the log id is a u16, which overflowed silently in release at 100k and panicked in debug. The breakdown shapes are now 10k, 30k, and 60k, and the illustrative numbers use the valid 50k batch. ANALYSIS.md and the README are updated with the streaming table and the u16 limit. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…aries The steady-state IPC saving looked schema-shaped because it is constant with batch size, but splitting the fixed per-batch amortization by Arrow IPC message type shows it is mostly dictionaries. Of the roughly 11.3 KB saved per steady-state batch at 1000 records with zstd, 7.7 KB are dictionary messages and 3.6 KB are schema, about two thirds dictionaries and one third schema. The cost is fixed with batch size because both parts are per-stream rather than per-row. The schema describes columns, and the set of distinct dictionary values does not grow with the row count in this synthetic data, so the dictionary messages do not grow either. The analysis also notes that this is a best case for dictionary amortization, because real high-cardinality columns such as a trace id or a log body carry new values every batch and their delta dictionaries do not amortize. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…ession) Assert that streaming batches after the second are identical in size, and document that Arrow IPC amortizes the schema and dictionary value tables once but does not compress batches against each other, so the per-row payload is re-sent every batch. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…ploited Add a measurement showing that Arrow IPC's per-batch independent compression stores near-duplicate streaming batches at full size (about 81 KB each at 10k logs), while a whole-stream zstd recovers the cross-batch redundancy only with a large window and long-distance matching (level 19 collapses eight near-duplicate batches to 69 KB, a marginal 269 bytes each), and a default-effort whole-stream zstd does not, because each uncompressed 2.4 MB batch exceeds its match window. - ipc.rs: new cross_batch_redundancy_needs_large_window test and helper. - ANALYSIS.md: new section with the measured table and three caveats (best-case duplicates, high CPU, loss of per-batch independent decode). - Cargo.toml: zstd dev-dependency for the whole-stream measurement. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Apply the read/write cost model to the deployment where one organization owns both the client exporter and the storage schema and is optimizing server-side ingestion CPU. Covers: the coupling objection dissolving into a layout-version surface, the store-versus-read hinge and Parquet footer/statistics enabling metadata-only routing without a full decode, moving transforms to the exporter while cross-source metric re-aggregation stays server-side, dual OTAP/IPC plus Parquet intake, and the zstd-versus-lz4 interaction on a .NET sender. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Parquet is written by arrow-rs in the Rust sender, so it always has zstd. The .NET receiver's Arrow reader supports both lz4 and zstd IPC decompression as of arrow-dotnet v23, but only via the separate Apache.Arrow.Compression package and its codec factory; the core package decompresses neither and there is no build where lz4 works but zstd does not. Split the consequence into the package-present case, where zstd IPC stays available, and the package-absent case, where IPC must travel uncompressed and most favors precomputed Parquet, and note that the table decode times are arrow-rs while the .NET receiver decode is a separate cost. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…ysis The .NET receiver's IPC compression support changes arrow-ipc performance but not the conclusion, so replace the two-case discussion with a single note that the Parquet is written by arrow-rs in the sender and the server-CPU argument does not depend on the compressor. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

The study is logs-only, so drop the cross-signal asides: replace the metric re-aggregation paragraph with a logs-specific version about per-record work moving to the gateway, and drop the trace-id example from the dictionary discussion. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…onversions Extend the otap_parquet study with an "OTAP-flat" intermediate: a single columnar view of OTAP logs that exploits the encoder emitting attributes grouped by parent_id, so the per-record log attributes are a zero-copy List<Struct> and only the shared resource/scope attributes vary by layout. - parquet_study::otap_flat: flatten/unflatten across three shared-attribute layouts (Materialized, RunEndEncoded, Dictionary), plus in_memory_bytes. - parquet_study::ipc_flat: plain Arrow-IPC serialize/read of the flat batch, which unlike Parquet carries run-end and dictionary columns on the wire. - parquet_study::parquet_io::to_parquet_ready: transform only the columns arrow-rs cannot round-trip (expand REE->List, materialize dict-FixedSizeBinary trace_id/span_id) and copy the rest. - datagen::RichGenParams: realistic high-cardinality data (unique trace ids, mixed-type attributes) so wire-size comparisons are not flattered by identical rows. - New printed tables in main.rs: OTAP-flat view cost/size, service-to-service transfer (ipc-standard vs ipc-flat vs parquet), and a per-edge conversion-cost matrix. - OTAP_FLAT_ANALYSIS.md: companion write-up positioning OTAP-flat as a natural center between OTAP-standard and Parquet, with a column-level copy-vs-transform analysis, sort-order notes, and the arrow-rs REE/dict-FixedSizeBinary limits. Benchmark/analysis only; not user-facing. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

… ordering precondition Recognizing the shared columns yields a direct OTAP-standard to Parquet path that skips the naive hash join: copy the Logs columns, attach log attributes zero-copy as List<Struct>, materialize resource/scope as lists, then write. The zero-copy log-attribute attach reads contiguous parent_id runs, so it is a correctness precondition, not just a perf tweak, that the attribute batches be grouped by parent_id. A freshly encoded batch is grouped; a transport-optimized one is sorted by (type, key, ...), which scatters parent_id. Verified with a probe: the fresh batch round-trips zero-copy while the wire-decoded batch fails the precondition and must be regrouped first. This splits the pipeline: a gateway that encodes OTLP to OTAP and writes Parquet in place gets the direct path free (precompute case), while a receiver of transport-optimized OTAP pays a regroup. Noted tension: transport optimization sorts Logs by (resource, scope, trace_id), which helps Parquet clustering, but sorts attributes by key, which the flatten must undo. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Add a table for the common OTLP -> batch -> Parquet case that times the naive hash-join flatten from ANALYSIS.md against the optimized shared-column zero-copy build, each followed by the same parquet-ready transform and Parquet write. Both produce byte-identical files, so only the flatten differs. The optimized flatten is several times cheaper on its own (log-heavy 47.7 -> 10.7 ms), but the Parquet writer is the shared floor, so the end-to-end saving is about 17% log-heavy and 6% resource-heavy. The gap reflects that the zero-copy build only helps the per-record log attributes; resource attributes must be materialized per row for Parquet on either path. Documented in OTAP_FLAT_ANALYSIS.md and README. Benchmark/analysis only; not user-facing. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Remove the .NET and SaaS references so the study is not tied to a specific vendor. The applied section now frames the motivating deployment generically: a single vendor controls both ends of the path, supplying the gateway software the customer runs on premises and operating the ingestion service, and because it owns both and wants to protect the ingestion service's CPU, it shifts the encoding cost onto the customer-run gateway. The compressor notes drop the ".NET builds" example and just say some Arrow/Parquet stacks may not support zstd. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Retitle the analysis docs as "part 1" and "part 2" with explicit reading order, share an identical units sentence, and make the cross-references consistent. Tighten wordy sections and consolidate the three overlapping closing sections of the flat analysis. No measurement tables changed. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Adds a new otap_parquet benchmark + supporting library code to study OTAP/IPC vs several flattened-Parquet encodings (nested/list, map, and wide/exploded), plus an OTAP-flat exploration and accompanying analysis writeups.

Changes:

Introduces the benchmarks::parquet_study module implementing multiple round-trippable codecs and utilities for flatten/unflatten, Parquet I/O, and dataset generation.
Adds the otap_parquet Criterion benchmark that prints multiple comparison tables (size, pipeline breakdowns, transfer, conversion matrix, etc.) and runs timed read/write benchmarks.
Adds benchmark documentation and analysis markdown describing results and implications.

Reviewed changes

Copilot reviewed 17 out of 17 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
rust/otap-dataflow/benchmarks/src/parquet_study/attrs.rs	Shared flatten/unflatten helpers, join keys, and round-trip structural assertions.
rust/otap-dataflow/benchmarks/src/parquet_study/datagen.rs	Synthetic OTLP→OTAP log batch generators for benchmark input shapes.
rust/otap-dataflow/benchmarks/src/parquet_study/ipc.rs	IPC baseline codec + sub-step timing helpers.
rust/otap-dataflow/benchmarks/src/parquet_study/ipc_flat.rs	In-memory Arrow IPC read/write for a single flat RecordBatch.
rust/otap-dataflow/benchmarks/src/parquet_study/map.rs	Parquet contender encoding attributes as `Map<Utf8, Struct{...}>`.
rust/otap-dataflow/benchmarks/src/parquet_study/mod.rs	Study scaffolding: `Codec`, `Compressor`, and scheme enumeration.
rust/otap-dataflow/benchmarks/src/parquet_study/nested.rs	Parquet contender encoding attributes as `List<Struct{key,...}>`.
rust/otap-dataflow/benchmarks/src/parquet_study/otap_flat.rs	OTAP-flat single-table views (materialized / REE / dictionary) + round-trip.
rust/otap-dataflow/benchmarks/src/parquet_study/parquet_io.rs	In-memory Parquet reader/writer plus “parquet-ready” materialization helper.
rust/otap-dataflow/benchmarks/src/parquet_study/server.rs	Server-side model helpers (IPC→Parquet conversion, Parquet reparse, persist).
rust/otap-dataflow/benchmarks/src/parquet_study/wide.rs	Wide “exploded key” Parquet contender with overflow for losslessness.
rust/otap-dataflow/benchmarks/benches/otap_parquet/main.rs	Criterion benchmark printing study tables and running timed read/write benches.
rust/otap-dataflow/benchmarks/benches/otap_parquet/README.md	Benchmark usage instructions and summarized findings.
rust/otap-dataflow/benchmarks/benches/otap_parquet/ANALYSIS.md	Part 1 write-up: OTAP/IPC vs flattened Parquet.
rust/otap-dataflow/benchmarks/benches/otap_parquet/OTAP_FLAT_ANALYSIS.md	Part 2 write-up: OTAP-flat single columnar view and conversions.
rust/otap-dataflow/benchmarks/src/lib.rs	Exposes `parquet_study` module from the benchmarks support library.
rust/otap-dataflow/benchmarks/Cargo.toml	Adds deps/features for Parquet study + registers the `otap_parquet` bench target.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+    debug_assert_eq!(
+        next_run,
+        run_lengths.len(),
+        "every LogAttrs run should map to exactly one log row with an id"
+    );


+            let keys: UInt16Array = (0..id_arr.len())
+                .map(|i| {
+                    let id = id_arr.value(i);
+                    u16::try_from(*pos_of_id.get(&id).unwrap_or(&0)).expect("key fits u16")
+                })
+                .collect();


+    fn write(&self, logs: OtapArrowRecords) -> StudyResult<Vec<u8>> {
+        let flat = flatten(&logs)?;
+        parquet_io::write_parquet(&flat, self.compressor.parquet())
+    }


+    fn write(&self, logs: OtapArrowRecords) -> StudyResult<Vec<u8>> {
+        let flat = flatten(&logs)?;
+        parquet_io::write_parquet(&flat, self.compressor.parquet())
+    }


+    fn write(&self, logs: OtapArrowRecords) -> StudyResult<Vec<u8>> {
+        let flat = flatten(&logs)?;
+        parquet_io::write_parquet(&flat, self.compressor.parquet())
+    }


+                let write_t = median_ms(|| {
+                    let _ = write_parquet(&flat, compressor.parquet()).expect("write");
+                });
+                let bytes = write_parquet(&flat, compressor.parquet()).expect("write");
+                let read_flat = read_parquet(&bytes).expect("read");


+            let pq = write_parquet(&flat, comp.parquet())
+                .expect("parquet write")
+                .len();


albertlockett · 2026-07-02T10:46:06Z

+`otap-flat-*` rows exploit the fact that OTAP attribute batches are already
+grouped by `parent_id`, so the per-record log attributes are a zero-copy
+`List<Struct>` and only the layout of the shared resource/scope sets differs. The


the fact that OTAP attribute batches are already grouped by parent_id

Not necessarily true - this is a happy accident of how OTAP is produced in most situations (e.g. when converting from OTLP), but I wouldn't advise relying on it systematically

albertlockett · 2026-07-02T10:52:46Z

+| S  -> Ws  standard serialize       |      62.9 |           25.8 |
+| Ws -> S   standard deserialize     |       9.9 |            4.5 |
+| F  -> Wf  flat serialize           |      37.0 |           12.2 |


I find the S -> Ws and F -> Wf numbers surprising here. Isn't this just doing arrow IPC serialization? That should be pretty cheap as its mostly just copying the arrow buffers.

albertlockett · 2026-07-02T10:53:23Z

+| zstd | 13.0  | 5.9     | 18.9    | 3.0     | 1.2   | 4.2     |
+| lz4  | 13.5  | 2.6     | 16.1    | 18.7    | 1.3   | 20.0    |
+| none | 12.6  | 29.6    | 42.2    | 27.1    | 1.2   | 28.3    |


Strange that the no-compression case performs the worst

I see that elsewhere we explain this by there being less data to copy.

I guess that makes sense because we are copying 400k vs 12mb into this vec here?

otel-arrow/rust/otap-dataflow/benchmarks/src/parquet_study/ipc.rs

Lines 43 to 51 in f74d711

pub fn encode_to_bytes(mut logs: OtapArrowRecords, compressor: Compressor) -> StudyResult<Vec<u8>> {

let mut producer = Producer::new_with_options(ProducerOptions {

ipc_compression: compressor.ipc(),

});

let bar = producer.produce_bar(&mut logs)?;

let mut buf = Vec::with_capacity(1024);

bar.encode(&mut buf)?;

Ok(buf)

}

FYI, the IPC writer also has an number of data copies, but there is work ongoing to resolve clean it up (apache/arrow-rs#10044 and apache/arrow-rs#10128)

albertlockett · 2026-07-02T11:29:21Z

+The OTAP encoder does not emit attributes in arbitrary order. It walks resources,
+scopes, and records in order, and for each parent it appends that parent's
+attributes contiguously before moving on. The result is that every attribute
+batch arrives already grouped by `parent_id` in ascending, contiguous runs, and
+the `Logs.id` column that keys the log attributes is a sequential, nullable
+`u16`. A probe over generated data confirms it directly. `ResourceAttrs.parent_id`
+is `0, 1, 2`, `ScopeAttrs.parent_id` is `0, 0, 1, 1, 2, 2`, and
+`LogAttrs.parent_id` is a run of zeros, then a run of ones, and so on.
+
+This ordering makes the hash join unnecessary. Two facts follow from it.
+
+First, the per-record log attributes can become a `List<Struct>` almost for free.
+The struct children are the existing `LogAttrs` value columns used as they are,
+with no `take`, and the list offsets come from a single linear scan that walks
+the sorted `parent_id` runs alongside the sequential `Logs.id` column. Every log
+attribute belongs to exactly one record, so no value is copied and no value is
+shared.
+
+Second, the shared resource and scope attributes line up with runs of log rows,
+because the records are already grouped by resource and then by scope. A resource
+therefore spans a contiguous block of rows, which is exactly the structure that
+run-end and dictionary encodings capture without physically repeating anything.


Although this is how the encoder works today, there's nothing that guarantees it has to work like that or that this sort-order needs to be maintained. Indeed, when we insert new attributes for example, we append them all to the end of the attrs record batch which breaks this assumption.

Also, if the attributes are were transport encoded, even if we remove the delta encoding we don't reorder the attribute table as far as I know. So again this assumption about the sort-order is incorrect.

FWIW when we proto-encode from OTAP, we have a similar problem where we need to group up attributes for some parent ID. In this case, we create these cursors for which rows to visit in order by sorting the ID columns.

otel-arrow/rust/otap-dataflow/crates/pdata/src/otlp/common.rs

Lines 1082 to 1240 in f74d711

/// Initializes the cursor to visit the root [`RecordBatch`] in the OTAP batch in order of

/// ascending Resource ID & Scope ID. This allows the caller to easily iterate the root

/// [`RecordBatch`] in a depth-first traversal for each Resource -> Scope -> Log/Span/etc.

pub fn init_cursor_for_root_batch(

&mut self,

record_batch: &RecordBatch,

cursor: &mut SortedBatchCursor,

) -> Result<()> {

let mut sort_columns_idx = 0;

let mut sort_columns: [Option<&UInt16Array>; 3] = [None, None, None];

for col_name in [consts::RESOURCE, consts::SCOPE] {

if let Some(struct_col) = record_batch.column_by_name(col_name) {

let struct_array = struct_col

.as_any()

.downcast_ref::<StructArray>()

.ok_or_else(|| Error::ColumnDataTypeMismatch {

name: col_name.into(),

expect: DataType::Struct(Fields::empty()),

actual: struct_col.data_type().clone(),

})?;

if let Some(ids) = struct_array.column_by_name(consts::ID) {

sort_columns[sort_columns_idx] =

Some(ids.as_any().downcast_ref::<UInt16Array>().ok_or_else(|| {

Error::ColumnDataTypeMismatch {

name: col_name.into(),

expect: DataType::UInt16,

actual: ids.data_type().clone(),

}

})?);

sort_columns_idx += 1;

}

}

}

if let Some(ids) = record_batch.column_by_name(consts::ID) {

sort_columns[sort_columns_idx] =

Some(ids.as_any().downcast_ref::<UInt16Array>().ok_or_else(|| {

Error::ColumnDataTypeMismatch {

name: consts::ID.into(),

expect: DataType::UInt16,

actual: ids.data_type().clone(),

}

})?);

}

match sort_columns {

// Pack 3 u16 IDs into a u64 for composite sorting by resource, scope, then row ID.

[Some(resource_ids), Some(scope_ids), Some(ids)] => {

self.u64_ids.clear();

self.u64_ids.extend(

resource_ids

.values()

.iter()

.zip(scope_ids.values().iter())

.zip(ids.values().iter())

.enumerate()

.map(|(i, ((r, s), id))| {

(i, (*r as u64) << 32 | (*s as u64) << 16 | *id as u64)

}),

);

self.u64_ids.sort_unstable_by_key(|&(_, v)| v);

cursor

.sorted_indices

.extend(self.u64_ids.iter().map(|(i, _)| *i));

}

// Pack 2 u16 IDs into a u32 for composite sorting by resource, then scope ID.

[Some(ids1), Some(ids2), None] => {

self.u32_ids.clear();

self.u32_ids.extend(

ids1.values()

.iter()

.zip(ids2.values().iter())

.enumerate()

.map(|(i, (a, b))| (i, (*a as u32) << 16 | *b as u32)),

);

self.u32_ids.sort_unstable_by_key(|&(_, v)| v);

cursor

.sorted_indices

.extend(self.u32_ids.iter().map(|(i, _)| *i));

}

// Single ID column — sort directly by u16 value.

[Some(ids), None, None] => {

self.init_cursor_for_u16_id_column(&MaybeDictArrayAccessor::Native(ids), cursor);

}

// No ID columns — visit in natural order.

[None, None, None] => {

cursor.sorted_indices.extend(0..record_batch.num_rows());

}

_ => unreachable!(),

}

Ok(())

}

pub fn init_cursor_for_u16_id_column(

&mut self,

ids: &MaybeDictArrayAccessor<'_, UInt16Array>,

cursor: &mut SortedBatchCursor,

) {

Self::init_cursor_for_ids_column(&mut self.u16_ids, ids, cursor);

}

pub fn init_cursor_for_u32_id_column(

&mut self,

ids: &MaybeDictArrayAccessor<'_, UInt32Array>,

cursor: &mut SortedBatchCursor,

) {

Self::init_cursor_for_ids_column(&mut self.u32_ids, ids, cursor);

}

fn init_cursor_for_ids_column<T: ArrowPrimitiveType>(

sort_ids_tmp: &mut Vec<(usize, T::Native)>,

ids: &MaybeDictArrayAccessor<'_, PrimitiveArray<T>>,

cursor: &mut SortedBatchCursor,

) where

<T as ArrowPrimitiveType>::Native: Ord,

{

sort_ids_tmp.clear();

match ids {

MaybeDictArrayAccessor::Native(ids) => {

sort_ids_tmp.extend(ids.values().iter().copied().enumerate());

}

MaybeDictArrayAccessor::Dictionary16(ids) => {

sort_ids_tmp.extend(

(0..ids.len())

.map(|i| ids.value_at(i).unwrap_or_default())

.enumerate(),

);

}

MaybeDictArrayAccessor::Dictionary8(ids) => {

sort_ids_tmp.extend(

(0..ids.len())

.map(|i| ids.value_at(i).unwrap_or_default())

.enumerate(),

);

}

}

if ids.null_count() == 0 {

// fast path, no null IDs

sort_ids_tmp.sort_unstable_by_key(|&(_, value)| value);

} else {

// sort nulls last

sort_ids_tmp.sort_unstable_by(|(ia, a), (ib, b)| {

match (ids.is_valid(*ia), ids.is_valid(*ib)) {

(true, true) => a.cmp(b),

(true, false) => Ordering::Less,

(false, true) => Ordering::Greater,

(false, false) => Ordering::Equal,

}

});

}

cursor

.sorted_indices

.extend(sort_ids_tmp.iter().map(|(i, _)| *i));

}

}

albertlockett · 2026-07-02T11:35:47Z

+Two arrow-rs 58.3 limits shape the results. The writer cannot serialize run-end
+or nested-dictionary columns, so those layouts are in-memory forms only, and the
+reader cannot read a dictionary-encoded `FixedSizeBinary` such as `trace_id`
+back, so the study materializes those columns before writing Parquet. Both may
+lift as arrow-rs adds support, at which point the run-end view could also become
+a Parquet write target.


REE support in parquet is being actively worked on apache/arrow-rs#10064

jmacd · 2026-07-02T15:00:26Z

I didn't mean to open this PR in this repo!!!!! It was to be in my fork, for saving.

jmacd · 2026-07-02T15:02:00Z

Thank you @albertlockett this was meant to be for discussion really really not ready to open for wide review. Thank you.

It leads me to believe we have a viable flat-Parquet story but also teaches me that Parquet is very expensive to write.

albertlockett · 2026-07-02T15:07:38Z

I didn't mean to open this PR in this repo!!!!! It was to be in my fork, for saving.

Thank you @albertlockett this was meant to be for discussion really really not ready to open for wide review. Thank you.

It leads me to believe we have a viable flat-Parquet story but also teaches me that Parquet is very expensive to write.

Happy to have reviewed - this was interesting to read!

I'm not too surprised - Parquet is quite aggressive about adding optimizing the storage of the columns. IIRC it tries to apply dictionary encoding + RLE + Bitpacking for most columns by default.

jmacd and others added 18 commits June 30, 2026 18:06

jmacd requested a review from a team as a code owner July 1, 2026 23:30

Copilot AI review requested due to automatic review settings July 1, 2026 23:30

github-project-automation Bot added this to OTel-Arrow Jul 1, 2026

github-actions Bot added lang:rust Pull requests that update Rust code size/XL labels Jul 1, 2026

Copilot started reviewing on behalf of jmacd July 1, 2026 23:30 View session

Copilot AI reviewed Jul 1, 2026

View reviewed changes

albertlockett reviewed Jul 2, 2026

View reviewed changes

jmacd closed this Jul 2, 2026

github-project-automation Bot moved this to Done in OTel-Arrow Jul 2, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

OTAP Parquet study - #3405

OTAP Parquet study#3405
jmacd wants to merge 18 commits into
open-telemetry:mainfrom
jmacd:jmacd/parquet_study

jmacd commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

albertlockett Jul 2, 2026

Uh oh!

jmacd commented Jul 2, 2026

Uh oh!

jmacd commented Jul 2, 2026

Uh oh!

albertlockett commented Jul 2, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	pub fn encode_to_bytes(mut logs: OtapArrowRecords, compressor: Compressor) -> StudyResult<Vec<u8>> {
	let mut producer = Producer::new_with_options(ProducerOptions {
	ipc_compression: compressor.ipc(),
	});
	let bar = producer.produce_bar(&mut logs)?;
	let mut buf = Vec::with_capacity(1024);
	bar.encode(&mut buf)?;
	Ok(buf)
	}

	/// Initializes the cursor to visit the root [`RecordBatch`] in the OTAP batch in order of
	/// ascending Resource ID & Scope ID. This allows the caller to easily iterate the root
	/// [`RecordBatch`] in a depth-first traversal for each Resource -> Scope -> Log/Span/etc.
	pub fn init_cursor_for_root_batch(
	&mut self,
	record_batch: &RecordBatch,
	cursor: &mut SortedBatchCursor,
	) -> Result<()> {
	let mut sort_columns_idx = 0;
	let mut sort_columns: [Option<&UInt16Array>; 3] = [None, None, None];
	for col_name in [consts::RESOURCE, consts::SCOPE] {
	if let Some(struct_col) = record_batch.column_by_name(col_name) {
	let struct_array = struct_col
	.as_any()
	.downcast_ref::<StructArray>()
	.ok_or_else(\|\| Error::ColumnDataTypeMismatch {
	name: col_name.into(),
	expect: DataType::Struct(Fields::empty()),
	actual: struct_col.data_type().clone(),
	})?;
	if let Some(ids) = struct_array.column_by_name(consts::ID) {
	sort_columns[sort_columns_idx] =
	Some(ids.as_any().downcast_ref::<UInt16Array>().ok_or_else(\|\| {
	Error::ColumnDataTypeMismatch {
	name: col_name.into(),
	expect: DataType::UInt16,
	actual: ids.data_type().clone(),
	}
	})?);
	sort_columns_idx += 1;
	}
	}
	}

	if let Some(ids) = record_batch.column_by_name(consts::ID) {
	sort_columns[sort_columns_idx] =
	Some(ids.as_any().downcast_ref::<UInt16Array>().ok_or_else(\|\| {
	Error::ColumnDataTypeMismatch {
	name: consts::ID.into(),
	expect: DataType::UInt16,
	actual: ids.data_type().clone(),
	}
	})?);
	}

	match sort_columns {
	// Pack 3 u16 IDs into a u64 for composite sorting by resource, scope, then row ID.
	[Some(resource_ids), Some(scope_ids), Some(ids)] => {
	self.u64_ids.clear();
	self.u64_ids.extend(
	resource_ids
	.values()
	.iter()
	.zip(scope_ids.values().iter())
	.zip(ids.values().iter())
	.enumerate()
	.map(\|(i, ((r, s), id))\| {
	(i, (r as u64) << 32 \| (s as u64) << 16 \| *id as u64)
	}),
	);
	self.u64_ids.sort_unstable_by_key(\|&(_, v)\| v);
	cursor
	.sorted_indices
	.extend(self.u64_ids.iter().map(\|(i, _)\| *i));
	}
	// Pack 2 u16 IDs into a u32 for composite sorting by resource, then scope ID.
	[Some(ids1), Some(ids2), None] => {
	self.u32_ids.clear();
	self.u32_ids.extend(
	ids1.values()
	.iter()
	.zip(ids2.values().iter())
	.enumerate()
	.map(\|(i, (a, b))\| (i, (a as u32) << 16 \| b as u32)),
	);
	self.u32_ids.sort_unstable_by_key(\|&(_, v)\| v);
	cursor
	.sorted_indices
	.extend(self.u32_ids.iter().map(\|(i, _)\| *i));
	}
	// Single ID column — sort directly by u16 value.
	[Some(ids), None, None] => {
	self.init_cursor_for_u16_id_column(&MaybeDictArrayAccessor::Native(ids), cursor);
	}
	// No ID columns — visit in natural order.
	[None, None, None] => {
	cursor.sorted_indices.extend(0..record_batch.num_rows());
	}
	_ => unreachable!(),
	}

	Ok(())
	}

	pub fn init_cursor_for_u16_id_column(
	&mut self,
	ids: &MaybeDictArrayAccessor<'_, UInt16Array>,
	cursor: &mut SortedBatchCursor,
	) {
	Self::init_cursor_for_ids_column(&mut self.u16_ids, ids, cursor);
	}

	pub fn init_cursor_for_u32_id_column(
	&mut self,
	ids: &MaybeDictArrayAccessor<'_, UInt32Array>,
	cursor: &mut SortedBatchCursor,
	) {
	Self::init_cursor_for_ids_column(&mut self.u32_ids, ids, cursor);
	}

	fn init_cursor_for_ids_column<T: ArrowPrimitiveType>(
	sort_ids_tmp: &mut Vec<(usize, T::Native)>,
	ids: &MaybeDictArrayAccessor<'_, PrimitiveArray<T>>,
	cursor: &mut SortedBatchCursor,
	) where
	<T as ArrowPrimitiveType>::Native: Ord,
	{
	sort_ids_tmp.clear();

	match ids {
	MaybeDictArrayAccessor::Native(ids) => {
	sort_ids_tmp.extend(ids.values().iter().copied().enumerate());
	}
	MaybeDictArrayAccessor::Dictionary16(ids) => {
	sort_ids_tmp.extend(
	(0..ids.len())
	.map(\|i\| ids.value_at(i).unwrap_or_default())
	.enumerate(),
	);
	}
	MaybeDictArrayAccessor::Dictionary8(ids) => {
	sort_ids_tmp.extend(
	(0..ids.len())
	.map(\|i\| ids.value_at(i).unwrap_or_default())
	.enumerate(),
	);
	}
	}

	if ids.null_count() == 0 {
	// fast path, no null IDs
	sort_ids_tmp.sort_unstable_by_key(\|&(_, value)\| value);
	} else {
	// sort nulls last
	sort_ids_tmp.sort_unstable_by(\|(ia, a), (ib, b)\| {
	match (ids.is_valid(ia), ids.is_valid(ib)) {
	(true, true) => a.cmp(b),
	(true, false) => Ordering::Less,
	(false, true) => Ordering::Greater,
	(false, false) => Ordering::Equal,
	}
	});
	}

	cursor
	.sorted_indices
	.extend(sort_ids_tmp.iter().map(\|(i, _)\| *i));
	}
	}

Uh oh!

Conversation

jmacd commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

albertlockett Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

jmacd commented Jul 2, 2026

Uh oh!

jmacd commented Jul 2, 2026

Uh oh!

albertlockett commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

albertlockett commented Jul 2, 2026 •

edited

Loading